package com.easyim.imserver.handler.message;

import cn.hutool.core.lang.Snowflake;
import com.alibaba.fastjson2.JSONObject;
import com.easyim.common.message.Immessage;
import io.netty.channel.Channel;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.util.Map;

import static com.easyim.imserver.constant.CacheKey.AUTHENTICATED;
import static com.easyim.imserver.constant.CacheKey.DATA_MSG_HANDLER_KEY;
import static com.easyim.imserver.utils.RedisKeyUtils.buildSessionMsgKey;
import static com.easyim.imserver.utils.RedisKeyUtils.buildUserSessionKey;

@Component
@Data
@Slf4j
public class DataMsgHandler implements Handler{

    @Autowired
    private Snowflake snowflake;

    @Autowired
    @Qualifier("channelMap")
    Map<String, Map<String, Channel>>  localSession;


    @Autowired
    @Qualifier("msgRedisTemplate")
    private RedisTemplate redisTemplate;

    private String machinedId;

    @Value("${com.easyim.run.model.clusterable:false}")
    private Boolean cluster;

    @Value("${com.easyim.message.kafka.sendMsgTopic}")
    private String sendTopic;

    @Autowired
    @Qualifier("msgSendKafkaTemplate")
    private KafkaTemplate<String,String> kafkaTemplate;

    @Override
    public String dealType() {
        return DATA_MSG_HANDLER_KEY;
    }

    @Override
    public void handlerMsg(Channel channel,
            Immessage.ImMessage imMessage) {
        //生成消息id 客户端会有个消息id 但是不能用 在不能用的情况下，依然要将消息id传回去
        Boolean isAuth = channel.attr(AUTHENTICATED).get();
        Immessage.DataMessage dataMessage = imMessage.getDataMessage();
        //找到用户的
        if (isAuth != null && isAuth) {
            log.info("reviser user msg from client:{}", dataMessage);
            Boolean isGroup = dataMessage.getChatType().getNumber() == 1;
            //存储消息
            Immessage.DataMessage hasIdMsg = storageMsg(dataMessage, isGroup);
            //本机投递
            localSendMsg(hasIdMsg.getReceiver(), hasIdMsg);
            //向客户端返回ack
            //其他服务器投递
            if(cluster){
                //集群版需要投递其他服务器

            }
            //写入ack
        } else {
            log.error("user not login, close connect:{}", dataMessage);
        }
    }


    private void localSendMsg(String toUsername, Immessage.DataMessage hasIdMsg){
        if(localSession.containsKey(toUsername)){
            //本台机器上 这个人是在线的 直接投递
            Map<String, Channel> deviceChannelMap = localSession.get(toUsername);
            deviceChannelMap.forEach(
                    (device, channel) -> {
                        //把消息推过去
                        if(channel.isActive()){
                            Immessage.ImMessage build =
                                    Immessage.ImMessage.newBuilder().setDataMessage(hasIdMsg)
                                            .build();
                            channel.writeAndFlush(build);
                        }else {
                            //todo 清理脏的会话
                        }
                    }
            );
        }
    }


    /**
     *  存储消息
     */
    private Immessage.DataMessage storageMsg(Immessage.DataMessage dataMessage, Boolean isGroup){
        long time = System.currentTimeMillis();
        //先发送到kafka 配置clickhouse消费kafka 然后查询消息的时候 从clickhouse拉取
        String sessionId = getSessionId(isGroup, dataMessage.getSender(), dataMessage.getReceiver());
        long msgId = snowflake.nextId();
        Immessage.DataMessage hasIdMsg = dataMessage.toBuilder().setServerMsgId(msgId)
                .setSessionId(sessionId)
                .setTimestamp(time).build();
        String s = JSONObject.toJSONString(hasIdMsg);
        kafkaTemplate.send(sendTopic, dataMessage.getSender(), s);
        //先更新用户会话 会话列表 zset结构 会话 如果单聊的话 就是两个用户的id拼接起来的 如果是群组 那么就是群组的id sessionKey msgId source是时间戳
        String sender = hasIdMsg.getSender();
        String receiver = hasIdMsg.getReceiver();

        redisTemplate.opsForZSet().add(buildSessionMsgKey(sessionId), msgId, time);
        //要更新用户的会话 必须两个的都更新 如果是单聊 群聊的话 只更新一个即可了
        redisTemplate.opsForZSet().add(buildUserSessionKey(sender), sessionId, time);
        if(!isGroup){
            redisTemplate.opsForZSet().add(buildUserSessionKey(receiver), sessionId, time);
        }
        return hasIdMsg;
    }

    private String getSessionId(Boolean isGroup, String sender, String receiver){
        StringBuilder chatSessionName = new StringBuilder();
        if(isGroup){
            //群聊由群id组成
            chatSessionName.append(receiver);
            return chatSessionName.toString();
        }else {
            //单聊由两个id组成
            if(sender.compareTo(receiver) < 0){
                chatSessionName.append(sender);
                chatSessionName.append("_");
                chatSessionName.append(receiver);
            }else {
                chatSessionName.append(receiver);
                chatSessionName.append("_");
                chatSessionName.append(sender);
            }
            return chatSessionName.toString();
        }
    }


}
